package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime.app.func.DorisSinkFunction;
import com.atguigu.edu.realtime.beans.DwsInteractionCourseCommentWindowBean;
import com.atguigu.edu.realtime.beans.DwsTradeCourseOrderWindowBean;
import com.atguigu.edu.realtime.utils.DateFormatUtil;
import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * 互动域课程评价指标统计
 */
public class DwsInteractionCourseCommentWindow {
    public static void main(String[] args) {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(5);
        //TODO 2.检查点相关的设置
        env.enableCheckpointing(5000L);
        //TODO 3.从kakfa主题中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "dwd_interaction_comment_info";
        String groupId = "dws_interaction_course_comment_window";
        //3.2 创建消费者对象
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        //3.3 消费数据 封装为流
        DataStreamSource<String> kafkaStrDS
                = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");

        //TODO 4.对流中数据进行类型转换  jsonStr->jsonObj->实体类对象
        SingleOutputStreamOperator<DwsInteractionCourseCommentWindowBean> beanStreamDS
        = kafkaStrDS.process(
         new ProcessFunction<String, DwsInteractionCourseCommentWindowBean>() {
            @Override
            public void processElement(String value, Context context, Collector<DwsInteractionCourseCommentWindowBean> out) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(value);
                String courseId = jsonObject.getString("course_id");
                Long reviewStars = jsonObject.getLong("review_stars");
                Long ts = jsonObject.getLong("ts");
                out.collect(DwsInteractionCourseCommentWindowBean.builder()
                        .courseId(courseId)
                        .reviewTotalStars(reviewStars)
                        .reviewUserCount(1L)
                        .goodReviewUserCount(reviewStars == 5L ? 1L : 0L)
                        .ts(ts)
                        .build());
             }
         }
    );
     // beanStreamDS.print(">>>");
     //TODO 5.设置水位线
        SingleOutputStreamOperator<DwsInteractionCourseCommentWindowBean> withWaterMarkStream
        = beanStreamDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<DwsInteractionCourseCommentWindowBean>forBoundedOutOfOrderness(Duration.ofSeconds(5L))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<DwsInteractionCourseCommentWindowBean>() {
                                    @Override
                                    public long extractTimestamp(DwsInteractionCourseCommentWindowBean element, long l) {
                                        return element.getTs();
                                    }
                                }
                        )
        );

      //TODO 6.按照课程ID分组
        KeyedStream<DwsInteractionCourseCommentWindowBean, String> courseKeyDS
                = withWaterMarkStream.keyBy(
                new KeySelector<DwsInteractionCourseCommentWindowBean, String>() {
                    @Override
                    public String getKey(DwsInteractionCourseCommentWindowBean value) throws Exception {
                        return value.getCourseId();
                    }
                }
        );
        //courseKeyDS.print("###");
        //TODO 7.开窗聚合
        SingleOutputStreamOperator<DwsInteractionCourseCommentWindowBean> reduceStream = courseKeyDS.window(
                 TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(2L)))
                .reduce(new ReduceFunction<DwsInteractionCourseCommentWindowBean>() {
                    @Override
                    public DwsInteractionCourseCommentWindowBean reduce(DwsInteractionCourseCommentWindowBean value1, DwsInteractionCourseCommentWindowBean value2) throws Exception {
                        value1.setReviewTotalStars(value1.getReviewTotalStars() + value2.getReviewTotalStars());
                        value1.setGoodReviewUserCount(value1.getGoodReviewUserCount()+value2.getGoodReviewUserCount());
                        value1.setReviewUserCount(value1.getReviewUserCount()+ value2.getReviewUserCount());
                        //System.out.println(value1);
                        return value1;
                    }
                }, new WindowFunction<DwsInteractionCourseCommentWindowBean, DwsInteractionCourseCommentWindowBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<DwsInteractionCourseCommentWindowBean> input, Collector<DwsInteractionCourseCommentWindowBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(window.getStart());
                        String edt = DateFormatUtil.toYmdHms(window.getEnd());
                        String curDate = DateFormatUtil.toDate(window.getStart());
                        for (DwsInteractionCourseCommentWindowBean orderBean : input) {
                            orderBean.setStt(stt);
                            orderBean.setEdt(edt);
                            orderBean.setCurDate(curDate);
                            out.collect(orderBean);
                        }
                    }
                });
         reduceStream.print("^^^^");
        /*//TODO 8.维度关联课程信息
        SingleOutputStreamOperator<DwsInteractionCourseCommentWindowBean> dimBeanDS
                = AsyncDataStream.unorderedWait(
                reduceStream,
                new DimAsyncFunction<DwsInteractionCourseCommentWindowBean>("dim_course_info") {
                    @Override
                    public void join(JSONObject JsonObj, DwsInteractionCourseCommentWindowBean bean) {
                     bean.setCourseName(JsonObj.getString("course_name"));
                    }

                    @Override
                    public String getKey(DwsInteractionCourseCommentWindowBean obj) {
                        return obj.getCourseId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        //TODO 9.写入到Doris中
        dimBeanDS.print("****");
        dimBeanDS.addSink(new DorisSinkFunction<DwsInteractionCourseCommentWindowBean>("dws_interaction_course_comment_window"));
*/

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
